﻿/**
 //*code is far away from bug with the animal protecting
 //*  ┏┓　　　┏┓
 //*┏┛┻━━━┛┻┓
 //*┃　　　　　　　┃ 　
 //*┃　　　━　　　┃
 //*┃　┳┛　┗┳　┃
 //*┃　　　　　　　┃
 //*┃　　　┻　　　┃
 //*┃　　　　　　　┃
 //*┗━┓　　　┏━┛
 //*　　┃　　　┃神兽保佑
 //*　　┃　　　┃代码无BUG！
 //*　　┃　　　┗━━━┓
 //*　　┃　　　　　　　┣┓
 //*　　┃　　　　　　　┏┛
 //*　　┗┓┓┏━┳┓┏┛
 //*　　　┃┫┫　┃┫┫
 //*　　　┗┻┛　┗┻┛
 //*　　
 //*  
 //*   ---------------------------------
 //*      @author (shang)
 //*      @since  	2021/05/08 19:26:46
 */
using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Text.Json;
namespace KafkaService.Implements
{
    public class KafkaService : IKafkaService
    {
        public async Task PublishAsync<TMessage>(string topicName, TMessage message) where TMessage : class
        {
            var config = new ProducerConfig
            {
                BootstrapServers = "192.168.8.110:9092"
            };
            using var producer = new ProducerBuilder<string, string>(config).Build();
            await producer.ProduceAsync(topicName, new Message<string, string>
            {
                Key = Guid.NewGuid().ToString(),
                Value =JsonSerializer.Serialize(message)
            });
        }

        public async Task SubscribeAsync<TMessage>(IEnumerable<string> topics, Action<TMessage> messageFunc, CancellationToken cancellationToken) where TMessage : class
        {
            var config = new ConsumerConfig
            {
                BootstrapServers = "192.168.8.110:9092",
                //groupID:
                // 一个字符串用来指示一组consumer所在的组。相同的groupID表示在一个组里。相同的groupID消费记录offset时，记录的是同一个offset
                //。
                // 所以，此处需要注意，（1）如果多个地方都使用相同的groupid，可能造成个别消费者消费不到的情况（2）如果单个消费者消费能力不足的话，可以启动多个相同groupid的consumer消费，处理相同的逻辑。但是，多线程的时候，需要增加每个groupid下的partition分区数量，便于每个线程稳定读取固定的partition，提高消费能力。
                GroupId = "crow-consumer",
                EnableAutoCommit = false,
                StatisticsIntervalMs = 5000,
                SessionTimeoutMs = 6000,
                //offset：用来保存消费进度。
                //offset表示在当前topic，当前groupID下消费到的位置。
                //offset为earliest并不代表offset=1.
                //在不进行过期配置的情况下，
                //kafka消息默认7天时间就会过期。
                //过期后其offset也就随之发生变化，
                //使得用数字进行配置的消费进度并不准确。
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnablePartitionEof = true
            };
            using var consumer = new ConsumerBuilder<Ignore, string>(config)
              .SetErrorHandler((_, e) =>
              {
                  Console.WriteLine($"Error: {e.Reason}");
              })
              .SetStatisticsHandler((_, json) =>
              {
                  Console.WriteLine($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} > 消息监听中..");
              })
              .SetPartitionsAssignedHandler((c, partitions) =>
              {
                  string partitionsStr = string.Join(", ", partitions);
                  Console.WriteLine($" - 分配的 kafka 分区: {partitionsStr}");
              })
              .SetPartitionsRevokedHandler((c, partitions) =>
              {
                  string partitionsStr = string.Join(", ", partitions);
                  Console.WriteLine($" - 回收了 kafka 的分区: {partitionsStr}");
              })
              .Build();
             consumer.Subscribe(topics);

            try
            {
                while (true)
                {
                    try
                    {
                        var consumeResult = consumer.Consume(cancellationToken);
                        Console.WriteLine($"Consumed message '{consumeResult.Message?.Value}' at: '{consumeResult?.TopicPartitionOffset}'.");
                        if (consumeResult.IsPartitionEOF)
                        {
                            Console.WriteLine($" - {DateTime.Now:yyyy-MM-dd HH:mm:ss} 已经到底了：{consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");
                            continue;
                        }
                        TMessage messageResult = null;
                        try
                        {
                            messageResult = JsonSerializer.Deserialize<TMessage>(consumeResult.Message.Value);
                        }
                        catch (Exception ex)
                        {
                            var errorMessage = $" - {DateTime.Now:yyyy-MM-dd HH:mm:ss}【Exception 消息反序列化失败，Value：{consumeResult.Message.Value}】 ：{ex.StackTrace?.ToString()}";
                            Console.WriteLine(errorMessage);
                            messageResult = null;
                        }
                        if (messageResult != null/* && consumeResult.Offset % commitPeriod == 0*/)
                        {
                            messageFunc(messageResult);
                            try
                            {
                                consumer.Commit(consumeResult);
                            }
                            catch (KafkaException e)
                            {
                                Console.WriteLine(e.Message);
                            }
                        }
                    }
                    catch (ConsumeException e)
                    {
                        Console.WriteLine($"Consume error: {e.Error.Reason}");
                    }
                }
            }
            catch (OperationCanceledException)
            {

                Console.WriteLine("Closing consumer.");
                consumer.Close();
            }
            await Task.CompletedTask;
        }
    }
}
